package com.shujia.flink.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

object Demo2SourceFunction {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    //使用自定义source
    val myDS: DataStream[Int] = env.addSource(new MySource)

    myDS.print()

    env.execute()


  }

}

/**
 * 自定义source,实现SourceFunction接口
 *
 */
class MySource extends SourceFunction[Int] {
  /**
   * run方法只执行一次
   * run用于读取数据方法，可以将读取到的数据发送到下游
   *
   * @param ctx ：用于将数据发送到下游
   */
  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {

    while (true) {
      //发送数据到下游
      ctx.collect(1)

      Thread.sleep(100)
    }
  }


  //用于再任务被取消时回收资源
  override def cancel(): Unit = {}
}
